package com.ibm.cps.message;

import java.io.Serializable;
import java.util.Collection;
import java.util.HashSet;
import java.util.Set;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.KryoSerializable;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.google.gson.Gson;
import com.google.gson.JsonArray;
import com.google.gson.JsonObject;
import com.google.gson.JsonParser;
import com.ibm.cps.message.MessageFactory.MessageType;

public class SlidingWindowMetaMessage implements KryoSerializable, Serializable {

	String processorId;
	Set<String> parentProcessors;
	String attributeName;
	int size;
	int step;
	String type;

	public SlidingWindowMetaMessage(String processorId,
			Set<String> parentProcessors, String attributeName, int size,
			int step) {
		this.processorId = processorId;
		this.parentProcessors = parentProcessors;
		this.attributeName = attributeName;
		this.size = size;
		this.step = step;
		this.type = MessageType.SLIDING_WINDOW.name();
	}

	/**
	 * json format
	 * tenantid schemaid topicName rulename ruleStr
	 * @param ruleString
	 */
	public SlidingWindowMetaMessage(byte[] sdByte) {
		String slidingWindow = new String(sdByte);
		System.out.println(slidingWindow);
		JsonParser parser = new JsonParser();
		JsonObject swObject = parser.parse(slidingWindow).getAsJsonObject();
		processorId = swObject.get("processorId").getAsString();
		attributeName = swObject.get("attributeName").getAsString();
		size = swObject.get("size").getAsInt();
		step = swObject.get("step").getAsInt();
		type = swObject.get("type").getAsString();
		JsonArray dsArray = swObject.get("parentProcessors").getAsJsonArray();
		parentProcessors = new HashSet<String>(dsArray.size());
		for (int i = 0; i < dsArray.size(); i++) {
			parentProcessors.add(dsArray.get(i).getAsString());
		}
	}

	public String getSWJsonStr() {
		Gson gson = new Gson();
		return gson.toJson(this);
	}

	@Override
	public void write(Kryo kryo, Output output) {
		output.writeString(processorId);
		output.writeString(attributeName);
		output.writeInt(size);
		output.writeInt(step);
		kryo.writeObject(output, parentProcessors);
	}

	@Override
	public void read(Kryo kryo, Input input) {
		processorId = input.readString();
		attributeName = input.readString();
		size = input.readInt();
		step = input.readInt();
		parentProcessors = kryo.readObject(input, Set.class);
	}

	public String getProcessorId() {
		return processorId;
	}

	public Set<String> getParentProcessors() {
		return parentProcessors;
	}

	public String getAttributeName() {
		return attributeName;
	}

	public int getSize() {
		return size;
	}

	public int getStep() {
		return step;
	}

	public static String getCollectionWindows(
			Collection<SlidingWindowMetaMessage> windows) {
		String jsonString = "{\"slinding_windows\": [";
		boolean isFirst = true;
		for (SlidingWindowMetaMessage window : windows) {
			if (!isFirst) {
				jsonString += ",";
			}
			jsonString += window.getSWJsonStr();
			isFirst = false;
		}
		jsonString += "]}";
		return jsonString;
	}

}
